package com.ouyangtao.spark.session;

import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Optional;
import com.ouyangtao.constant.Constants;
import com.ouyangtao.dao.*;
import com.ouyangtao.dao.factory.DAOFactory;
import com.ouyangtao.domain.*;
import com.ouyangtao.util.*;
import it.unimi.dsi.fastutil.ints.IntArrayList;
import it.unimi.dsi.fastutil.ints.IntList;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.storage.StorageLevel;
import scala.Tuple2;

import java.util.*;

/**
 * 用户访问session分析spark作业
 * @author taoouyang
 */
public class UserVisitSessionAnalyzeSpark {

    public static void main(String[] args) {
        //构建上下文
        SparkConf conf = new SparkConf()
                .setAppName(Constants.SPARK_APP_NAME_SESSION)
//              .set("spark.default.parallelism","100")
                .set("spark.storage.memoryFraction", "0.5")
                .set("spark.shuffle.consolidateFiles", "true")
                .set("spark.shuffle.file.buffer", "64")
                .set("spark.shuffle.memoryFraction", "0.3")
                .set("spark.reducer.maxSizeInFlight", "24")
                .set("spark.shuffle.io.maxRetries", "60")
                .set("spark.shuffle.io.retryWait", "60")
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .registerKryoClasses(new Class[]{CategorySortKey.class,IntList.class});
        SparkUtils.setMaster(conf);
        /**
         * 比如，获取top10热门品类功能中，二次排序，自定义了一个Key
         * 那个key是需要在进行shuffle的时候，进行网络传输的，因此也是要求实现序列化的
         * 启用Kryo机制以后，就会用Kryo去序列化和反序列化CategorySortKey
         * 所以这里要求，为了获取最佳性能，注册一下我们自定义的类
         */
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = SparkUtils.getSQLContext(sc.sc());

        //生成模拟测试数
        SparkUtils.mockData(sc,sqlContext);

        //创建需要使用的DAO组件
        ITaskDAO taskDao = DAOFactory.getTaskDAO();

        //首先得查询出来指定的任务,并获取任务的查询参数
        long taskId = ParamUtils.getTaskIdFromArgs(args,Constants.SPARK_LOCAL_TASKID_SESSION);
        Task task = taskDao.findById(taskId);
        if(task == null) {
            System.out.println(new Date() + ": cannot find this task with id [" + taskId + "].");
            return;
        }

        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());

        //如果要进行session粒度的数据聚合,要从user_visit_action表中,查询出来自动日期范围内的行为数据
        /**
         * actionRDD，就是一个公共RDD
         * 第一，要用actionRDD，获取到一个公共的sessionId为key的PairRDD
         * 第二，actionRDD，用在了session聚合环节里面
         *
         * sessionId为key的PairRDD，是确定了，在后面要多次使用的
         * 1、与通过筛选的sessionId进行join，获取通过筛选的session的明细数据
         * 2、将这个RDD，直接传入aggregateBySession方法，进行session聚合统计
         *
         * 重构完以后，actionRDD，就只在最开始，使用一次，用来生成以sessionId为key的RDD
         *
         */
        JavaRDD<Row> actionRDD = SparkUtils.getActionRDDByDateRange(sqlContext,taskParam);
        JavaPairRDD<String,Row> sessionId2ActionRDD = getSessionId2ActionRDD(actionRDD);

        /**
         * 持久化，很简单，就是对RDD调用persist()方法，并传入一个持久化级别
         *
         * 如果是persist(StorageLevel.MEMORY_ONLY())，纯内存，无序列化，那么就可以用cache()方法来替代
         *
         * 如果内存充足，要使用双副本高可靠机制
         * 选择后缀带_2的策略
         * StorageLevel.MEMORY_ONLY_2()
         *
         */
        sessionId2ActionRDD = sessionId2ActionRDD.persist(StorageLevel.MEMORY_ONLY());

        // 首先，可以将行为数据，按照session_id进行groupByKey分组
        // 此时的数据的粒度就是session粒度了，然后呢，可以将session粒度的数据
        // 与用户信息数据，进行join
        // 然后就可以获取到session粒度的数据，同时呢，数据里面还包含了session对应的user的信息
        JavaPairRDD<String,String> sessionID2AggInfoRDD = aggregateBySession(sc,sqlContext,sessionId2ActionRDD);

        // 接着，就要针对session粒度的聚合数据，按照使用者指定的筛选参数进行数据过滤
        // 相当于我们自己编写的算子，是要访问外面的任务参数对象的
        // 所以，大家记得我们之前说的，匿名内部类（算子函数），访问外部对象，是要给外部对象使用final修饰的
        //重构,同时进行过滤和统计
        Accumulator<String> sessionAggStatAccumulator = sc.accumulator("",new SessionAggStatAccumulator());

        JavaPairRDD<String, String> filteredSessionId2AggInfoRDD =
                filterSessionAndAggStat(sessionID2AggInfoRDD, taskParam,sessionAggStatAccumulator);

       filteredSessionId2AggInfoRDD =  filteredSessionId2AggInfoRDD.persist(StorageLevel.MEMORY_ONLY());
        //生成公共的RDD,通过筛选条件的session的访问明细数据
        /**
         * 重构：sessionId2detailRDD，就是代表了通过筛选的session对应的访问明细数据
         */
        JavaPairRDD<String,Row> sessionId2DetailRDD = getSessionId2DetailRDD(filteredSessionId2AggInfoRDD,sessionId2ActionRDD);

        sessionId2DetailRDD = sessionId2DetailRDD.persist(StorageLevel.MEMORY_ONLY());

        //必须把能够触发job执行的操作(某一个action操作之后)放在最终写入mysql方法之前
       randomExtractSession(sc,task.getTaskId(),filteredSessionId2AggInfoRDD,sessionId2ActionRDD);

       // filteredSessionId2AggInfoRDD.count();
        //计算出各个范围的session占比,并写入mysql
        calculateAndPersistAggStat(sessionAggStatAccumulator.value(),task.getTaskId());

        //获取top10热门品类
        List<Tuple2<CategorySortKey, String>> top10CategoryList = getTop10Category(taskId, sessionId2DetailRDD);

        //获取top10活跃session
        getTop10Session(sc,task.getTaskId(),top10CategoryList,sessionId2DetailRDD);

        //关闭上下文
        sc.close();

    }

    /**
     * 获取sessionId2到访问新闻数据的映射的RDD
     * @param actionRDD
     * @return
     */
    private static JavaPairRDD<String,Row> getSessionId2ActionRDD(JavaRDD<Row> actionRDD) {

       /* return actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Row> call(Row row) throws Exception {
                return new Tuple2<String, Row>(row.getString(2),row);
            }
        });*/
       return actionRDD.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Row>, String, Row>() {

           private static final long serialVersionUID = 1L;
           @Override
           public Iterable<Tuple2<String, Row>> call(Iterator<Row> iterator) throws Exception {
              List<Tuple2<String,Row>> list = new ArrayList<>();

              while (iterator.hasNext()){
                  Row row = iterator.next();
                  list.add(new Tuple2<String, Row>(row.getString(2),row));
              }
               return list;
           }
       });
    }

    /**
     * 对行为数据按session粒度进行聚合
     * @param sqlContext
     * @param sessionId2ActionRDD
     * @return session粒度聚合数据
     */
    private static JavaPairRDD<String,String> aggregateBySession(JavaSparkContext sc,SQLContext sqlContext,  JavaPairRDD<String,Row> sessionId2ActionRDD){


        //对行为数据按session粒度进行分组
        JavaPairRDD<String,Iterable<Row>> sessionId2ActionsRDD = sessionId2ActionRDD.groupByKey();

        //对每一个session分组进行聚合,将session中所有的搜索词和点击品类都聚合起来
        // 到此为止，获取的数据格式，如下：<userId,partAggInfo(sessionId,searchKeywords,clickCategoryIds)>
        JavaPairRDD<Long, String> userId2PartAggInfoRDD = sessionId2ActionsRDD.mapToPair(

                new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {

                        String sessionId = tuple._1;
                        Iterator<Row> iterator = tuple._2.iterator();

                        StringBuffer searchKeywordsBuffer = new StringBuffer("");
                        StringBuffer clickCategoryIdsBuffer = new StringBuffer("");

                        Long userId = null;

                        //session起始和结束时间
                        Date startTime = null;
                        Date endTime = null;

                        //session访问步长
                        int stepLength = 0;

                        //遍历session所有的访问行为
                        while (iterator.hasNext()) {
                            //提取每个访问行为的搜索词字段和点击品类字段
                            Row row = iterator.next();
                            if (userId==null){
                                userId = row.getLong(1);
                            }
                            String searchKeyword = row.getString(5);
                            Long clickCategoryId = row.getLong(6);
                            // 实际上这里要对数据说明一下
                            // 并不是每一行访问行为都有searchKeyword何clickCategoryId两个字段的
                            // 其实，只有搜索行为，是有searchKeyword字段的
                            // 只有点击品类的行为，是有clickCategoryId字段的
                            // 所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的

                            // 我们决定是否将搜索词或点击品类id拼接到字符串中去
                            // 首先要满足：不能是null值
                            // 其次，之前的字符串中还没有搜索词或者点击品类id
                            if (StringUtils.isNotEmpty(searchKeyword)){
                                if (!searchKeywordsBuffer.toString().contains(searchKeyword)){
                                    searchKeywordsBuffer.append(searchKeyword+",");
                                }
                            }

                            if (clickCategoryId!=null){
                                if (!clickCategoryIdsBuffer.toString().contains(String.valueOf(clickCategoryId))){
                                    clickCategoryIdsBuffer.append(clickCategoryId+",");
                                }
                            }

                            //计算session开始和结束时间
                            Date actionTime = DateUtils.parseTime(row.getString(4));

                            if (startTime == null){
                                startTime = actionTime;
                            }

                            if (endTime == null){
                                endTime = actionTime;
                            }

                            if (actionTime.before(startTime)){
                                startTime = actionTime;
                            }

                            if (actionTime.after(endTime)){
                                endTime = actionTime;
                            }

                            //计算session步长
                            stepLength++;

                        }

                        String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
                        String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());

                        //计算session访问时长
                        long visitLength = (endTime.getTime()-startTime.getTime())/1000;


                        // 大家思考一下
                        // 我们返回的数据格式，即使<sessionid,partAggrInfo>
                        // 但是，这一步聚合完了以后，其实，我们是还需要将每一行数据，跟对应的用户信息进行聚合
                        // 问题就来了，如果是跟用户信息进行聚合的话，那么key，就不应该是sessionid
                        // 就应该是userid，才能够跟<userid,Row>格式的用户信息进行聚合
                        // 如果我们这里直接返回<sessionid,partAggrInfo>，还得再做一次mapToPair算子
                        // 将RDD映射成<userid,partAggrInfo>的格式，那么就多此一举

                        // 所以，我们这里其实可以直接，返回的数据格式，就是<userid,partAggrInfo>
                        // 然后跟用户信息join的时候，将partAggrInfo关联上userInfo
                        // 然后再直接将返回的Tuple的key设置成sessionid
                        // 最后的数据格式，还是<sessionid,fullAggrInfo>

                        // 聚合数据，用什么样的格式进行拼接？
                        // 我们这里统一定义，使用key=value|key=value
                        String partAggInfo = Constants.FIELD_SESSION_ID + "=" + sessionId + "|"
                                + Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|"
                                + Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds + "|"
                                + Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|"
                                + Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|"
                                + Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime);
                        return new Tuple2<Long,String>(userId,partAggInfo);
                    }
                });

        // 查询所有用户数据，并映射成<userId,Row>的格式
        String sql = "select * from user_info";
        JavaRDD<Row> userInfoRDD = sqlContext.sql(sql).javaRDD();

        JavaPairRDD<Long, Row> userId2InfoRDD = userInfoRDD.mapToPair(
                new PairFunction<Row, Long, Row>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<Long, Row> call(Row row) throws Exception {
                        return new Tuple2<Long,Row>(row.getLong(0),row);
                    }
                });

        /**
         * 这里就可以说一下，比较适合采用reduce join转换为map join的方式
         *
         * userId2PartAggInfoRDD，可能数据量还是比较大，比如，可能在1千万数据
         * userId2InfoRDD，可能数据量还是比较小的，你的用户数量才10万用户
         *
         */

      /*  List<Tuple2<Long, Row>> userInfos = userId2InfoRDD.collect();
        final Broadcast<List<Tuple2<Long, Row>>> userInfosBroadcast = sc.broadcast(userInfos);
        JavaPairRDD<String, String> sessionId2FullAggInfoRDD = userId2PartAggInfoRDD.mapToPair(new PairFunction<Tuple2<Long, String>, String, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple) throws Exception {
                //得到用户信息map
                List<Tuple2<Long, Row>> userInfos = userInfosBroadcast.value();
                Map<Long,Row> userInfoMap = new HashMap<>();
                for (Tuple2<Long, Row> userInfo : userInfos) {
                    userInfoMap.put(userInfo._1,userInfo._2);
                }
                //获取到当前用户对应的信息
                String partAggInfo = tuple._2;
                Row userInfoRow = userInfoMap.get(tuple._1);

                String sessionId = StringUtils.getFieldFromConcatString(partAggInfo,"\\|",Constants.FIELD_SESSION_ID);

                int age = userInfoRow.getInt(3);
                String professional = userInfoRow.getString(4);
                String city = userInfoRow.getString(5);
                String sex = userInfoRow.getString(6);


                String fullAggInfo = partAggInfo + "|"
                        + Constants.FIELD_AGE + "=" + age + "|"
                        + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                        + Constants.FIELD_CITY + "=" + city + "|"
                        + Constants.FIELD_SEX + "=" + sex;


                return new Tuple2<String,String>(sessionId,fullAggInfo);
            }
        });
    */

        /**
         * sample采样倾斜key单独进行join
         */
      /*  JavaPairRDD<Long, String> sampleRDD = userId2PartAggInfoRDD.sample(false, 0.1, 9);

        JavaPairRDD<Long, Long> mappedSampledRDD = sampleRDD.mapToPair(new PairFunction<Tuple2<Long, String>, Long, Long>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, String> tuple) throws Exception {
                return new Tuple2<Long, Long>(tuple._1, 1L);
            }
        });

        JavaPairRDD<Long, Long> computedSampledRDD = mappedSampledRDD.reduceByKey(new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

        JavaPairRDD<Long, Long> reversedSampledRDD = computedSampledRDD.mapToPair(new PairFunction<Tuple2<Long, Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, Long> tuple) throws Exception {
                return new Tuple2<Long, Long>(tuple._2, tuple._1);
            }
        });

       final Long skewedUserId =  reversedSampledRDD.sortByKey(false).take(1).get(0)._2;

        JavaPairRDD<Long, String> skewedRDD = userId2PartAggInfoRDD.filter(new Function<Tuple2<Long, String>, Boolean>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return tuple._1.equals(skewedUserId);
            }
        });

        JavaPairRDD<Long, String> commonRDD = userId2PartAggInfoRDD.filter(new Function<Tuple2<Long, String>, Boolean>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return !tuple._1.equals(skewedUserId);
            }
        });

        JavaPairRDD<String, Row> skewedUserId2InfoRDD = userId2InfoRDD.filter(new Function<Tuple2<Long, Row>, Boolean>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Boolean call(Tuple2<Long, Row> tuple) throws Exception {
                return tuple._1.equals(skewedUserId);
            }
        }).flatMapToPair(new PairFlatMapFunction<Tuple2<Long, Row>, String, Row>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<String, Row>> call(Tuple2<Long, Row> tuple) throws Exception {
                Random random = new Random();
                List<Tuple2<String, Row>> list = new ArrayList<>();
                for (int i = 0; i < 100; i++) {
                    int prefix = random.nextInt(100);
                    list.add(new Tuple2<String, Row>(prefix + "_" + tuple._1, tuple._2));
                }
                return list;
            }
        });

        JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD1 = skewedRDD.mapToPair(new PairFunction<Tuple2<Long, String>, String, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple) throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(100);
                return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
            }
        }).join(skewedUserId2InfoRDD).mapToPair(new PairFunction<Tuple2<String, Tuple2<String, Row>>, Long, Tuple2<String, Row>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, Tuple2<String, Row>> call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
                long userId = Long.valueOf(tuple._1.split("_")[1]);
                return new Tuple2<Long, Tuple2<String, Row>>(userId, tuple._2);
            }
        });

        JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(userId2InfoRDD);

        JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD = joinedRDD1.union(joinedRDD2);

        JavaPairRDD<String, String> sessionId2FullAggInfoRDD = joinedRDD.mapToPair(new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tuple) throws Exception {
                String partAggInfo = tuple._2._1;
                Row userInfoRow = tuple._2._2;

                String sessionId = StringUtils.getFieldFromConcatString(
                        partAggInfo, "\\|", Constants.FIELD_SESSION_ID);

                int age = userInfoRow.getInt(3);
                String professional = userInfoRow.getString(4);
                String city = userInfoRow.getString(5);
                String sex = userInfoRow.getString(6);

                String fullAggInfo = partAggInfo + "|"
                        + Constants.FIELD_AGE + "=" + age + "|"
                        + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                        + Constants.FIELD_CITY + "=" + city + "|"
                        + Constants.FIELD_SEX + "=" + sex;

                return new Tuple2<String, String>(sessionId, fullAggInfo);
            }
        });
*/




        //将session粒度聚合数据与用户信息进行join
        JavaPairRDD<Long,Tuple2<String,Row>> userId2FullInfoRDD = userId2PartAggInfoRDD.join(userId2InfoRDD);


        // 对join起来的数据进行拼接，并且返回<sessionId,fullAggInfo>格式的数据
        JavaPairRDD<String,String> sessionId2FullAggInfoRDD = userId2FullInfoRDD.mapToPair(

                new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tuple) throws Exception {

                   String partAggInfo = tuple._2._1;
                   Row userInfoRow = tuple._2._2;
                   String sessionId = StringUtils.getFieldFromConcatString(partAggInfo,"\\|",Constants.FIELD_SESSION_ID);

                   int age = userInfoRow.getInt(3);
                   String professional = userInfoRow.getString(4);
                   String city = userInfoRow.getString(5);
                   String sex = userInfoRow.getString(6);


                   String fullAggInfo = partAggInfo + "|"
                        + Constants.FIELD_AGE + "=" + age + "|"
                        + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                        + Constants.FIELD_CITY + "=" + city + "|"
                        + Constants.FIELD_SEX + "=" + sex;


                    return new Tuple2<String,String>(sessionId,fullAggInfo);
                }
        });

        /**
         * 使用随机数和扩容表进行join
         */

 /*       JavaPairRDD<String, Row> expandedRDD = userId2InfoRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<Long, Row>, String, Row>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<String, Row>> call(Tuple2<Long, Row> tuple) throws Exception {
                List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
                for (int i = 0; i < 10; i++) {
                    list.add(new Tuple2<String, Row>(0 + "_" + tuple._1, tuple._2));
                }
                return list;
            }
        });

        JavaPairRDD<String, String> mappedRDD = userId2PartAggInfoRDD.mapToPair(new PairFunction<Tuple2<Long, String>, String, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple) throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(10);
                return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
            }
        });

        JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);
        JavaPairRDD<String, String> finalRDD = joinedRDD.mapToPair(new PairFunction<Tuple2<String, Tuple2<String, Row>>, String, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, String> call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
                String partAggInfo = tuple._2._1;
                Row userInfoRow = tuple._2._2;
                String sessionId = StringUtils.getFieldFromConcatString(partAggInfo,"\\|",Constants.FIELD_SESSION_ID);
                int age = userInfoRow.getInt(3);
                String professional = userInfoRow.getString(4);
                String city = userInfoRow.getString(5);
                String sex = userInfoRow.getString(6);
                String fullAggInfo = partAggInfo + "|"
                        + Constants.FIELD_AGE + "=" + age + "|"
                        + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                        + Constants.FIELD_CITY + "=" + city + "|"
                        + Constants.FIELD_SEX + "=" + sex;
                return new Tuple2<String,String>(sessionId,fullAggInfo);
            }
        });
*/

        return sessionId2FullAggInfoRDD;
    }

    /**
     * 过滤session数据
     * @param sessionId2AggInfoRDD
     * @param taskParam
     * @return
     */
    private static JavaPairRDD<String,String> filterSessionAndAggStat(JavaPairRDD<String,String> sessionId2AggInfoRDD,final JSONObject taskParam,final Accumulator<String> sessionAggStatAccumulator){

        // 为了使用我们后面的ValieUtils，所以，首先将所有的筛选参数拼接成一个连接串
        // 此外，这里其实大家不要觉得是多此一举
        // 其实我们是给后面的性能优化埋下了一个伏笔
        String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
        String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES);
        String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);

        String _parameter = (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
                + (endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "")
                + (professionals != null ? Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" : "")
                + (cities != null ? Constants.PARAM_CITIES + "=" + cities + "|" : "")
                + (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
                + (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
                + (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds: "");

        if (_parameter.endsWith("\\|")){
            _parameter = _parameter.substring(0,_parameter.length()-1);
        }

        final String parameter = _parameter;

        //根据筛选参数进行过滤
        JavaPairRDD<String,String> filteredSessionId2AggInfoRDD = sessionId2AggInfoRDD.filter(

                new Function<Tuple2<String, String>, Boolean>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Boolean call(Tuple2<String, String> tuple) throws Exception {
                    //从tuple中获取聚合数据
                        String aggInfo = tuple._2;
                        // 依次按照筛选条件进行过滤
                        // 按照年龄范围进行过滤（startAge、endAge）
                        if (!ValidUtils.between(aggInfo,Constants.FIELD_AGE,parameter,Constants.PARAM_START_AGE,Constants.PARAM_END_AGE)){
                            return false;
                        }

                        // 按照职业范围进行过滤（professionals）
                        // 互联网,IT,软件
                        // 互联网
                        if (!ValidUtils.in(aggInfo,Constants.FIELD_PROFESSIONAL,parameter,Constants.PARAM_PROFESSIONALS)){
                            return false;
                        }

                        // 按照城市范围进行过滤（cities）
                        // 北京,上海,广州,深圳
                        // 成都
                        if(!ValidUtils.in(aggInfo, Constants.FIELD_CITY,
                                parameter, Constants.PARAM_CITIES)) {
                            return false;
                        }
                        // 按照性别进行过滤
                        // 男/女
                        // 男，女
                        if(!ValidUtils.equal(aggInfo, Constants.FIELD_SEX,
                                parameter, Constants.PARAM_SEX)) {
                            return false;
                        }

                        // 按照搜索词进行过滤
                        // 我们的session可能搜索了 火锅,蛋糕,烧烤
                        // 我们的筛选条件可能是 火锅,串串香,iphone手机
                        // 那么，in这个校验方法，主要判定session搜索的词中，有任何一个，与筛选条件中
                        // 任何一个搜索词相当，即通过
                        if(!ValidUtils.in(aggInfo, Constants.FIELD_SEARCH_KEYWORDS,
                                parameter, Constants.PARAM_KEYWORDS)) {
                            return false;
                        }

                        // 按照点击品类id进行过滤
                        if(!ValidUtils.in(aggInfo, Constants.FIELD_CLICK_CATEGORY_IDS,
                                parameter, Constants.PARAM_CATEGORY_IDS)) {
                            return false;
                        }

                        // 如果经过了之前的多个过滤条件之后，程序能够走到这里
                        // 那么就说明，该session是通过了用户指定的筛选条件的，也就是需要保留的session
                        // 那么就要对session的访问时长和访问步长，进行统计，根据session对应的范围
                        // 进行相应的累加计数

                        // 主要走到这一步，那么就是需要计数的session
                        sessionAggStatAccumulator.add(Constants.SESSION_COUNT);

                        // 计算出session的访问时长和访问步长的范围，并进行相应的累加
                        long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(aggInfo,"\\|",Constants.FIELD_VISIT_LENGTH));
                        long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(aggInfo,"\\|",Constants.FIELD_STEP_LENGTH));
                        calculateVisitLength(visitLength);
                        calculateStepLength(stepLength);
                        return true;
                    }


                 /* * 计算访问时长范围
                * @param visitLength
                */
        private void calculateVisitLength(long visitLength) {
            if(visitLength >=1 && visitLength <= 3) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_1s_3s);
            } else if(visitLength >=4 && visitLength <= 6) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_4s_6s);
            } else if(visitLength >=7 && visitLength <= 9) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_7s_9s);
            } else if(visitLength >=10 && visitLength <= 30) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_10s_30s);
            } else if(visitLength > 30 && visitLength <= 60) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_30s_60s);
            } else if(visitLength > 60 && visitLength <= 180) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_1m_3m);
            } else if(visitLength > 180 && visitLength <= 600) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_3m_10m);
            } else if(visitLength > 600 && visitLength <= 1800) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_10m_30m);
            } else if(visitLength > 1800) {
                sessionAggStatAccumulator.add(Constants.TIME_PERIOD_30m);
            }
        }

        /**
         * 计算访问步长范围
         * @param stepLength
         */
        private void calculateStepLength(long stepLength) {
            if(stepLength >= 1 && stepLength <= 3) {
                sessionAggStatAccumulator.add(Constants.STEP_PERIOD_1_3);
            } else if(stepLength >= 4 && stepLength <= 6) {
                sessionAggStatAccumulator.add(Constants.STEP_PERIOD_4_6);
            } else if(stepLength >= 7 && stepLength <= 9) {
                sessionAggStatAccumulator.add(Constants.STEP_PERIOD_7_9);
            } else if(stepLength >= 10 && stepLength <= 30) {
                sessionAggStatAccumulator.add(Constants.STEP_PERIOD_10_30);
            } else if(stepLength > 30 && stepLength <= 60) {
                sessionAggStatAccumulator.add(Constants.STEP_PERIOD_30_60);
            } else if(stepLength > 60) {
                sessionAggStatAccumulator.add(Constants.STEP_PERIOD_60);
            }
        }

    });
        return filteredSessionId2AggInfoRDD;
    }

    /**
     * 获取通过筛选条件的session的访问明细数据RDD
     * @param sessionId2AggInfoRDD
     * @param sessionId2ActionRDD
     * @return
     */
    private static JavaPairRDD<String,Row> getSessionId2DetailRDD(JavaPairRDD<String,String> sessionId2AggInfoRDD,JavaPairRDD<String,Row> sessionId2ActionRDD){
        JavaPairRDD<String, Row> sessionId2DetailRDD = sessionId2AggInfoRDD
                .join(sessionId2ActionRDD)
                .mapToPair(new PairFunction<Tuple2<String, Tuple2<String, Row>>, String, Row>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Row> call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {

                        return new Tuple2<String, Row>(tuple._1, tuple._2._2);
                    }
                });

        return sessionId2DetailRDD;
    }

    /**
     * 随机抽取session
     * @param sessionId2AggInfoRDD
     */
    private static void randomExtractSession(JavaSparkContext sc,final long taskId,JavaPairRDD<String, String> sessionId2AggInfoRDD,JavaPairRDD<String,Row> sessionId2ActionRDD) {
        /**
         * 第一步计算出每天每小时的session数量,获取<yyyy-MM-dd_HH,sessionid>格式的RDD
        */
         JavaPairRDD<String, String> time2SessionIdRDD = sessionId2AggInfoRDD.mapToPair(
                new PairFunction<Tuple2<String, String>, String, String>() {

                    private static final long serialVersionUID = 1L;
                    @Override
                    public Tuple2<String, String> call(Tuple2<String, String> tuple) throws Exception {

                        String aggInfo = tuple._2;
                        String startTime = StringUtils.getFieldFromConcatString(aggInfo, "\\|", Constants.FIELD_START_TIME);
                        String dateHour = DateUtils.getDateHour(startTime);
                        return new Tuple2<String, String>(dateHour, aggInfo);
                    }
                });

        /* 每天每小时的session数量，然后计算出每天每小时的session抽取索引，遍历每天每小时session
                * 首先抽取出的session的聚合数据，写入session_random_extract表
                * 所以第一个RDD的value，应该是session聚合数据
                * 得到每天每小时的session数量
		 */

        //得到每天每小时的session数量,是有可能出现数据倾斜
         Map<String,Object> countMap = time2SessionIdRDD.countByKey();

        /*第二步使用按时间比例随机抽取算法,计算出每天每小时要抽取session的索引
         将<yyyy-MM-dd_HH,count>格式的map，转换成<yyyy-MM-dd,<HH,count>>的格式
        */
        Map<String,Map<String,Long>> dateHourCountMap = new HashMap<>();

        for (Map.Entry<String,Object> countEntry: countMap.entrySet()) {

            String dateHour = countEntry.getKey();
            String date = dateHour.split("_")[0];
            String hour = dateHour.split("_")[1];
            long count = Long.valueOf(String.valueOf(countEntry.getValue()));

            Map<String, Long> hourCountMap = dateHourCountMap.get(date);
            if(hourCountMap == null) {
                hourCountMap = new HashMap<String, Long>();
                dateHourCountMap.put(date, hourCountMap);
            }
            hourCountMap.put(hour, count);
        }

        //开始实现按时间比例随机抽取算法

        //总共要抽取100个session,先按照天数进行平分
        //dateHourCountMap为<"2019-10-22",<"01",15>>
        int extractNumberPerDay = 100 / dateHourCountMap.size();

        //比如 <date,<hour,(1,5,8,17这些索引)>>
        final Map<String,Map<String,List<Integer>>> dateHourExtractMap = new HashMap<>();
        Random random = new Random();

       for (Map.Entry<String,Map<String,Long>>dateHourCountEntry: dateHourCountMap.entrySet()){
           String date = dateHourCountEntry.getKey();
           Map<String,Long> hourCountMap = dateHourCountEntry.getValue();

           //计算出这一天的session总数
           long sessionCount = 0L;
           for (long hourCount:hourCountMap.values()) {
               sessionCount += hourCount;
           }

           Map<String,List<Integer>> hourExtractMap = dateHourExtractMap.get(date);
           if (hourExtractMap == null){
               hourExtractMap = new HashMap<String,List<Integer>>();
               dateHourExtractMap.put(date,hourExtractMap);
           }

           //遍历每小时
           for (Map.Entry<String,Long> hourCountEntry:hourCountMap.entrySet()) {
               String hour = hourCountEntry.getKey();
               long count = hourCountEntry.getValue();
               //计算每个小时的session数量占据当天总session数量比例,直接乘以每天要抽取的数量
               int hourExtractNumber = (int)(((double)count/(double)sessionCount)*extractNumberPerDay);
               if (hourExtractNumber>count){
                   hourExtractNumber = (int) count;
               }

               //先获取当前小时的存放随机数的list
               List<Integer> extractIndexList = hourExtractMap.get(hour);
               if (extractIndexList == null){
                   extractIndexList = new ArrayList<>();
                   hourExtractMap.put(hour,extractIndexList);
               }

               //生成上面计算出来的数量随机数
               for (int i = 0; i < hourExtractNumber; i++) {
                   int extractIndex = random.nextInt((int)count);
                   while (extractIndexList.contains(extractIndex)){
                       extractIndex = random.nextInt((int)count);
                   }
                   extractIndexList.add(extractIndex);
               }
           }
       }

        /**
         * fastUtil的使用，很简单，比如List<Integer>的list，对应到fastUtil，就是IntList
         */
        Map<String,Map<String,IntList>> fastUtilDateHourExtractMap = new HashMap<>();

        for (Map.Entry<String, Map<String, List<Integer>>> dateHourExtractEntry : dateHourExtractMap.entrySet()) {

            String date = dateHourExtractEntry.getKey();
            Map<String,List<Integer>> hourExtractMap = dateHourExtractEntry.getValue();
            Map<String,IntList> fastUtilHourExtractMap = new HashMap<>();
            for (Map.Entry<String, List<Integer>> hourExtractEntry : hourExtractMap.entrySet()) {
                String hour = hourExtractEntry.getKey();
                List<Integer> extractList = hourExtractEntry.getValue();

                IntList fastUtilExtractList = new IntArrayList();
                int extractListSize = extractList.size();
                for (int i = 0; i < extractListSize; i++) {
                    fastUtilExtractList.add(extractList.get(i));
                }
                fastUtilHourExtractMap.put(hour,fastUtilExtractList);
            }

            fastUtilDateHourExtractMap.put(date,fastUtilHourExtractMap);
        }



        /**
         * 广播变量，很简单
         * 其实就是SparkContext的broadcast()方法，传入你要广播的变量，即可
         */
        final Broadcast<Map<String,Map<String,IntList>>> dateHourExtractMapBroadcast = sc.broadcast(fastUtilDateHourExtractMap);


        /**
         * 第三步:遍历每天每小时的session,然后根据随机索引进行抽取
         */

        // 执行groupByKey算子，得到<dateHour,(session aggInfo)>
        JavaPairRDD<String, Iterable<String>> time2SessionsRDD = time2SessionIdRDD.groupByKey();

        // 我们用flatMap算子，遍历所有的<dateHour,(session aggInfo)>格式的数据
        // 然后呢，会遍历每天每小时的session
        // 如果发现某个session恰巧在我们指定的这天这小时的随机抽取索引上
        // 那么抽取该session，直接写入MySQL的random_extract_session表
        // 将抽取出来的session id返回回来，形成一个新的JavaRDD<String>
        // 然后最后一步，是用抽取出来的sessionId，去join它们的访问行为明细数据，写入session表

        JavaPairRDD<String, String> extractSessionIdsRDD = time2SessionsRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Iterable<String>>, String, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<String, String>> call(Tuple2<String, Iterable<String>> tuple) throws Exception {

                List<Tuple2<String,String>> extractSessionIds = new ArrayList<>();

                String dateHour = tuple._1;
                String date = dateHour.split("_")[0];
                String hour = dateHour.split("_")[1];
                Iterator<String> iterator = tuple._2.iterator();

                /**
                 * 使用广播变量的时候
                 * 直接调用广播变量的value/getValue()
                 * 可以获取到之前封装的广播变量
                 */
                Map<String,Map<String,IntList>> dateHourExtractMap = dateHourExtractMapBroadcast.getValue();
                List<Integer> extractIndexList = dateHourExtractMap.get(date).get(hour);

                ISessionRandomExtractDAO sessionRandomExtractDAO = DAOFactory.getSessionRandomExtractDAO();
                int index = 0;
                while (iterator.hasNext()){
                    String sessionAggInfo = iterator.next();

                    if (extractIndexList.contains(index)){
                        String sessionId = StringUtils.getFieldFromConcatString(sessionAggInfo,"\\|",Constants.FIELD_SESSION_ID);
                        // 将数据写入MySQL
                        SessionRandomExtract sessionRandomExtract = new SessionRandomExtract();
                        sessionRandomExtract.setTaskId(taskId);
                        sessionRandomExtract.setSessionId(sessionId);
                        sessionRandomExtract.setStartTime(StringUtils.getFieldFromConcatString(
                                sessionAggInfo, "\\|", Constants.FIELD_START_TIME));
                        sessionRandomExtract.setSearchKeywords(StringUtils.getFieldFromConcatString(
                                sessionAggInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS));
                        sessionRandomExtract.setClickCategoryIds(StringUtils.getFieldFromConcatString(
                                sessionAggInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS));

                        sessionRandomExtractDAO.insert(sessionRandomExtract);

                        //将sessionId加入list
                        extractSessionIds.add(new Tuple2<String, String>(sessionId,sessionId));

                    }
                    index++;
                }

                return extractSessionIds;
            }
        });

        /**
         * 第四步:获取抽取出来的session的明细数据
         */
        JavaPairRDD<String, Tuple2<String, Row>> extractSessionDetailRDD = extractSessionIdsRDD.join(sessionId2ActionRDD);

        extractSessionDetailRDD.foreach(new VoidFunction<Tuple2<String, Tuple2<String, Row>>>() {

            private static final long serialVersionUID = 1L;
            @Override
            public void call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {

                Row row = tuple._2._2;

                SessionDetail sessionDetail = new SessionDetail();
                sessionDetail.setTaskId(taskId);
                sessionDetail.setUserId(row.getLong(1));
                sessionDetail.setSessionId(row.getString(2));
                sessionDetail.setPageId(row.getLong(3));
                sessionDetail.setActionTime(row.getString(4));
                sessionDetail.setSearchKeyword(row.getString(5));
                sessionDetail.setClickCategoryId(row.getLong(6));
                sessionDetail.setClickProductId(row.getLong(7));
                sessionDetail.setOrderCategoryIds(row.getString(8));
                sessionDetail.setOrderProductIds(row.getString(9));
                sessionDetail.setPayCategoryIds(row.getString(10));
                sessionDetail.setPayProductIds(row.getString(11));

                ISessionDetailDAO sessionDetailDAO = DAOFactory.getSessionDetailDAO();
                sessionDetailDAO.insert(sessionDetail);

            }
        });

    }

    /**
     * 计算各session范围占比,并写入mysql
     * @param value
     * @param taskId
     */
    private static void calculateAndPersistAggStat(String value, long taskId) {
        //从accumulator统计串中获取值
        long session_count = Long.valueOf(StringUtils.getFieldFromConcatString(value,"\\|",Constants.SESSION_COUNT));

        long visit_length_1s_3s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1s_3s));
        long visit_length_4s_6s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_4s_6s));
        long visit_length_7s_9s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_7s_9s));
        long visit_length_10s_30s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10s_30s));
        long visit_length_30s_60s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30s_60s));
        long visit_length_1m_3m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1m_3m));
        long visit_length_3m_10m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_3m_10m));
        long visit_length_10m_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10m_30m));
        long visit_length_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30m));

        long step_length_1_3 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_1_3));
        long step_length_4_6 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_4_6));
        long step_length_7_9 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_7_9));
        long step_length_10_30 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_10_30));
        long step_length_30_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_30_60));
        long step_length_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_60));

        // 计算各个访问时长和访问步长的范围
        double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
                (double)visit_length_1s_3s / (double)session_count, 2);
        double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
                (double)visit_length_4s_6s / (double)session_count, 2);
        double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
                (double)visit_length_7s_9s / (double)session_count, 2);
        double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
                (double)visit_length_10s_30s / (double)session_count, 2);
        double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
                (double)visit_length_30s_60s / (double)session_count, 2);
        double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
                (double)visit_length_1m_3m / (double)session_count, 2);
        double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
                (double)visit_length_3m_10m / (double)session_count, 2);
        double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
                (double)visit_length_10m_30m / (double)session_count, 2);
        double visit_length_30m_ratio = NumberUtils.formatDouble(
                (double)visit_length_30m / (double)session_count, 2);

        double step_length_1_3_ratio = NumberUtils.formatDouble(
                (double)step_length_1_3 / (double)session_count, 2);
        double step_length_4_6_ratio = NumberUtils.formatDouble(
                (double)step_length_4_6 / (double)session_count, 2);
        double step_length_7_9_ratio = NumberUtils.formatDouble(
                (double)step_length_7_9 / (double)session_count, 2);
        double step_length_10_30_ratio = NumberUtils.formatDouble(
                (double)step_length_10_30 / (double)session_count, 2);
        double step_length_30_60_ratio = NumberUtils.formatDouble(
                (double)step_length_30_60 / (double)session_count, 2);
        double step_length_60_ratio = NumberUtils.formatDouble(
                (double)step_length_60 / (double)session_count, 2);

        // 将统计结果封装为Domain对象
        SessionAggStat sessionAggStat = new SessionAggStat();
        sessionAggStat.setTaskId(taskId);
        sessionAggStat.setSession_count(session_count);
        sessionAggStat.setVisit_length_1s_3s_ratio(visit_length_1s_3s_ratio);
        sessionAggStat.setVisit_length_4s_6s_ratio(visit_length_4s_6s_ratio);
        sessionAggStat.setVisit_length_7s_9s_ratio(visit_length_7s_9s_ratio);
        sessionAggStat.setVisit_length_10s_30s_ratio(visit_length_10s_30s_ratio);
        sessionAggStat.setVisit_length_30s_60s_ratio(visit_length_30s_60s_ratio);
        sessionAggStat.setVisit_length_1m_3m_ratio(visit_length_1m_3m_ratio);
        sessionAggStat.setVisit_length_3m_10m_ratio(visit_length_3m_10m_ratio);
        sessionAggStat.setVisit_length_10m_30m_ratio(visit_length_10m_30m_ratio);
        sessionAggStat.setVisit_length_30m_ratio(visit_length_30m_ratio);
        sessionAggStat.setStep_length_1_3_ratio(step_length_1_3_ratio);
        sessionAggStat.setStep_length_4_6_ratio(step_length_4_6_ratio);
        sessionAggStat.setStep_length_7_9_ratio(step_length_7_9_ratio);
        sessionAggStat.setStep_length_10_30_ratio(step_length_10_30_ratio);
        sessionAggStat.setStep_length_30_60_ratio(step_length_30_60_ratio);
        sessionAggStat.setStep_length_60_ratio(step_length_60_ratio);

        // 调用对应的DAO插入统计结果
        ISessionAggStatDAO sessionAggStatDAO = DAOFactory.getSessionAggStatDAO();
        sessionAggStatDAO.insert(sessionAggStat);
    }

    /**
     * 获取top10热门品类
     * @param sessionId2DetailRDD
     * @return 
     */
    private static List<Tuple2<CategorySortKey,String>> getTop10Category(long taskId, JavaPairRDD<String, Row> sessionId2DetailRDD) {
        /**
         * 第一步:获取符合条件的session访问过的所有品类
         */
        
        //获取session访问过(点击过,下单过,支付过)的所有品类id
        JavaPairRDD<Long, Long> categoryIdRDD = sessionId2DetailRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
                Row row = tuple._2;

                List<Tuple2<Long, Long>> list = new ArrayList<>();

                Long clickCategoryId = row.getLong(6);
                if (clickCategoryId != null) {
                    list.add(new Tuple2<Long, Long>(clickCategoryId, clickCategoryId));
                }
                String orderCategoryIds = row.getString(8);
                if (orderCategoryIds != null) {
                    String[] orderCategoryIdsSplit = orderCategoryIds.split(",");
                    for (String orderCategoryId : orderCategoryIdsSplit) {
                        list.add(new Tuple2<Long, Long>(Long.valueOf(orderCategoryId), Long.valueOf(orderCategoryId)));
                    }
                }

                String payCategoryIds = row.getString(10);
                if (payCategoryIds != null) {
                    String[] payCategoryIdsSplit = payCategoryIds.split(",");
                    for (String payCategoryId : payCategoryIdsSplit) {
                        list.add(new Tuple2<Long, Long>(Long.valueOf(payCategoryId),
                                Long.valueOf(payCategoryId)));
                    }
                }

                return list;
            }
        });

        /**
         * 必须要进行去重
         */
        categoryIdRDD = categoryIdRDD.distinct();

        /**
         * 第二步：计算各品类的点击、下单和支付的次数
         */

        // 访问明细中，其中三种访问行为是：点击、下单和支付
        // 分别来计算各品类点击、下单和支付的次数，可以先对访问明细数据进行过滤
        // 分别过滤出点击、下单和支付行为，然后通过map、reduceByKey等算子来进行计算

        //计算各个品类的点击次数
        JavaPairRDD<Long,Long> clickCategoryId2CountRDD = getClickCategoryId2CountRDD(sessionId2DetailRDD);
        // 计算各个品类的下单次数
        JavaPairRDD<Long, Long> orderCategoryId2CountRDD = getOrderCategoryId2CountRDD(sessionId2DetailRDD);
        // 计算各个品类的支付次数
        JavaPairRDD<Long, Long> payCategoryId2CountRDD = getPayCategoryId2CountRDD(sessionId2DetailRDD);

        /**
         * 第三步：join各品类与它的点击、下单和支付的次数
         *
         * categoryidRDD中，是包含了所有的符合条件的session，访问过的品类id
         *
         * 上面分别计算出来的三份，各品类的点击、下单和支付的次数，可能不是包含所有品类的
         * 比如，有的品类，就只是被点击过，但是没有人下单和支付
         *
         * 所以，这里，就不能使用join操作，要使用leftOuterJoin操作，就是说，如果categoryidRDD不能
         * join到自己的某个数据，比如点击、或下单、或支付次数，那么该categoryidRDD还是要保留下来的
         * 只不过，没有join到的那个数据，就是0了
         *
         */
       JavaPairRDD<Long,String> categoryId2CountRDD =  joinCategoryAndData(categoryIdRDD,clickCategoryId2CountRDD,orderCategoryId2CountRDD,payCategoryId2CountRDD);

        /**
         * 第四步：自定义二次排序key
         */

        /**
         * 第五步：将数据映射成<CategorySortKey,info>格式的RDD，然后进行二次排序（降序）
         */
        JavaPairRDD<CategorySortKey, String> sortKey2CountRDD = categoryId2CountRDD.mapToPair(new PairFunction<Tuple2<Long, String>, CategorySortKey, String>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<CategorySortKey, String> call(Tuple2<Long, String> tuple) throws Exception {
                String countInfo = tuple._2;
                long clickCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                        countInfo, "\\|", Constants.FIELD_CLICK_COUNT));
                long orderCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                        countInfo, "\\|", Constants.FIELD_ORDER_COUNT));
                long payCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                        countInfo, "\\|", Constants.FIELD_PAY_COUNT));

                CategorySortKey sortKey = new CategorySortKey(clickCount,
                        orderCount, payCount);
                return new Tuple2<CategorySortKey, String>(sortKey, countInfo);
            }
        });

        JavaPairRDD<CategorySortKey, String> sortedCategoryCountRDD =   sortKey2CountRDD.sortByKey(false);

        /**
         * 第六步：用take(10)取出top10热门品类，并写入MySQL
         */
        ITop10CategoryDAO top10CategoryDAO = DAOFactory.getTop10CategoryDAO();
        List<Tuple2<CategorySortKey, String>> top10CategoryList = sortedCategoryCountRDD.take(10);
        for (Tuple2<CategorySortKey, String> tuple : top10CategoryList) {
                String countInfo = tuple._2;
            long categoryId = Long.valueOf(StringUtils.getFieldFromConcatString(
                    countInfo, "\\|", Constants.FIELD_CATEGORY_ID));
            long clickCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                    countInfo, "\\|", Constants.FIELD_CLICK_COUNT));
            long orderCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                    countInfo, "\\|", Constants.FIELD_ORDER_COUNT));
            long payCount = Long.valueOf(StringUtils.getFieldFromConcatString(
                    countInfo, "\\|", Constants.FIELD_PAY_COUNT));

            Top10Category top10Category = new Top10Category();
            top10Category.setTaskId(taskId);
            top10Category.setCategoryId(categoryId);
            top10Category.setClickCount(clickCount);
            top10Category.setOrderCount(orderCount);
            top10Category.setPayCount(payCount);
            top10CategoryDAO.insert(top10Category);

        }

        return top10CategoryList;
    }



    /**
     * 获取各品类点击次数RDD
     * @param sessionId2DetailRDD
     * @return
     */
    private static JavaPairRDD<Long,Long> getClickCategoryId2CountRDD(JavaPairRDD<String, Row> sessionId2DetailRDD) {

        JavaPairRDD<String, Row> clickActionRDD = sessionId2DetailRDD.filter(new Function<Tuple2<String, Row>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<String, Row> tuple) throws Exception {
                Row row = tuple._2;
                return row.get(6) != null ? true : false;
            }
        });

        JavaPairRDD<Long, Long> clickCategoryIdRDD = clickActionRDD.mapToPair(new PairFunction<Tuple2<String, Row>, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, Long> call(Tuple2<String, Row> tuple) throws Exception {
                long clickCategoryId = tuple._2.getLong(6);
                return new Tuple2<Long, Long>(clickCategoryId, 1L);
            }
        });

        JavaPairRDD<Long, Long> clickCategoryId2CountRDD = clickCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

        /**
         * 提升shuffle reduce端并行度
         */
       /* JavaPairRDD<Long, Long> clickCategoryId2CountRDD = clickCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        },1000);*/


        /**
         * 使用随机key实现双重聚合
         */

    /*     // 第一步 ,给每个key打上一个随机数
        JavaPairRDD<String, Long> mapperClickCategoryIdRDD = clickCategoryIdRDD.mapToPair(new PairFunction<Tuple2<Long, Long>, String, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, Long> call(Tuple2<Long, Long> tuple) throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(10);
                return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);

            }
        });

        //第二步,执行第一步局部聚合
        JavaPairRDD<String, Long> firstAggRDD = mapperClickCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });


        //第三步,去除每个key的前缀
        JavaPairRDD<Long, Long> restoredRDD = firstAggRDD.mapToPair(new PairFunction<Tuple2<String, Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, Long> call(Tuple2<String, Long> tuple) throws Exception {
                long categoryId = Long.valueOf(tuple._1.split("_")[1]);
                return new Tuple2<Long, Long>(categoryId, tuple._2);
            }
        });

        //第四步,全局的聚合
        clickCategoryId2CountRDD =  restoredRDD.reduceByKey(new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1+v2;
            }
        });
*/

        return clickCategoryId2CountRDD;
    }

    /**
     * 获取各品类的下单次数RDD
     * @param sessionId2DetailRDD
     * @return
     */
    private static JavaPairRDD<Long, Long> getOrderCategoryId2CountRDD(
            JavaPairRDD<String, Row> sessionId2DetailRDD) {

        JavaPairRDD<String, Row> orderActionRDD = sessionId2DetailRDD.filter(new Function<Tuple2<String, Row>, Boolean>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Boolean call(Tuple2<String, Row> tuple) throws Exception {
                Row row = tuple._2;
                return row.getString(8) != null ? true : false;
            }
        });

        JavaPairRDD<Long, Long> orderCategoryIdRDD = orderActionRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
                Row row = tuple._2;
                String orderCategoryIds = row.getString(8);
                String[] orderCategoryIdsSplit = orderCategoryIds.split(",");
                List<Tuple2<Long, Long>> list = new ArrayList<>();
                for (String orderCategoryId : orderCategoryIdsSplit) {
                    list.add(new Tuple2<Long, Long>(Long.valueOf(orderCategoryId), 1L));
                }
                return list;
            }
        });

        JavaPairRDD<Long, Long> orderCategoryId2CountRDD = orderCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

        return orderCategoryId2CountRDD;
    }

  /**
     * 获取各品类的支付次数RDD
     * @param sessionId2DetailRDD
     * @return
     */
    private static JavaPairRDD<Long, Long> getPayCategoryId2CountRDD(
            JavaPairRDD<String, Row> sessionId2DetailRDD) {

        JavaPairRDD<String, Row> payActionRDD = sessionId2DetailRDD.filter(new Function<Tuple2<String, Row>, Boolean>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Boolean call(Tuple2<String, Row> tuple) throws Exception {
                Row row = tuple._2;
                return row.getString(10) != null ? true : false;
            }
        });

        JavaPairRDD<Long, Long> payCategoryIdRDD = payActionRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Row>, Long, Long>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<Long, Long>> call(Tuple2<String, Row> tuple) throws Exception {
                Row row = tuple._2;
                String payCategoryIds = row.getString(10);
                String[] payCategoryIdsSplit = payCategoryIds.split(",");
                List<Tuple2<Long, Long>> list = new ArrayList<>();
                for (String payCategoryId : payCategoryIdsSplit) {
                    list.add(new Tuple2<Long, Long>(Long.valueOf(payCategoryId), 1L));
                }
                return list;
            }
        });

        JavaPairRDD<Long, Long> payCategoryId2CountRDD = payCategoryIdRDD.reduceByKey(new Function2<Long, Long, Long>() {
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

        return payCategoryId2CountRDD;
    }
    /**
     * 连接品类RDD与数据RDD
     * @param categoryIdRDD
     * @param clickCategoryId2CountRDD
     * @param orderCategoryId2CountRDD
     * @param payCategoryId2CountRDD
     * @return
     */
    private static JavaPairRDD<Long,String> joinCategoryAndData(JavaPairRDD<Long, Long> categoryIdRDD, final JavaPairRDD<Long, Long> clickCategoryId2CountRDD, JavaPairRDD<Long, Long> orderCategoryId2CountRDD, JavaPairRDD<Long, Long> payCategoryId2CountRDD) {
        JavaPairRDD<Long, Tuple2<Long, Optional<Long>>> tmpJoinRDD = categoryIdRDD.leftOuterJoin(clickCategoryId2CountRDD);

        JavaPairRDD<Long, String> tmpMapRDD = tmpJoinRDD.mapToPair(new PairFunction<Tuple2<Long, Tuple2<Long, Optional<Long>>>, Long, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, String> call(Tuple2<Long, Tuple2<Long, Optional<Long>>> tuple) throws Exception {
                long categoryId = tuple._1;
                Optional<Long> optional = tuple._2._2;
                long clickCount = 0L;
                if (optional.isPresent()) {
                    clickCount = optional.get();
                }

                String value = Constants.FIELD_CATEGORY_ID + "=" + categoryId + "|" + Constants.FIELD_CLICK_COUNT + "=" + clickCount;

                return new Tuple2<Long, String>(categoryId, value);
            }
        });

        tmpMapRDD = tmpMapRDD.leftOuterJoin(orderCategoryId2CountRDD).mapToPair(new PairFunction<Tuple2<Long, Tuple2<String, Optional<Long>>>, Long, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, String> call(Tuple2<Long, Tuple2<String, Optional<Long>>> tuple) throws Exception {
                long categoryId = tuple._1;
                String value = tuple._2._1;
                Optional<Long> optional = tuple._2._2;
                long orderCount = 0L;
                if (optional.isPresent()) {
                    orderCount = optional.get();
                }
                 value = value + "|" + Constants.FIELD_ORDER_COUNT + "=" + orderCount;

                return new Tuple2<Long, String>(categoryId, value);
            }
        });

        tmpMapRDD = tmpMapRDD.leftOuterJoin(payCategoryId2CountRDD).mapToPair(new PairFunction<Tuple2<Long, Tuple2<String, Optional<Long>>>, Long, String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, String> call(Tuple2<Long, Tuple2<String, Optional<Long>>> tuple) throws Exception {
                long categoryId = tuple._1;
                String value = tuple._2._1;
                Optional<Long> optional = tuple._2._2;
                long payCount = 0L;
                if (optional.isPresent()) {
                    payCount = optional.get();
                }
                value = value + "|" + Constants.FIELD_PAY_COUNT + "=" + payCount;

                return new Tuple2<Long, String>(categoryId, value);
            }
        });

        return tmpMapRDD;
    }

    /**
     * 获取top10活跃的session
     * @param sc
     * @param taskId
     * @param top10CategoryList
     * @param sessionId2DetailRDD
     */
    private static void getTop10Session(JavaSparkContext sc, final long taskId, List<Tuple2<CategorySortKey, String>> top10CategoryList, JavaPairRDD<String, Row> sessionId2DetailRDD) {

        //第一步:将top10热门品类的id生成一份RDD
        List<Tuple2<Long,Long>> top10CategoryIdList = new ArrayList<>();

        for (Tuple2<CategorySortKey, String> category : top10CategoryList) {
            long categoryId = Long.valueOf(StringUtils.getFieldFromConcatString(category._2,"\\|",Constants.FIELD_CATEGORY_ID));
            top10CategoryIdList.add(new Tuple2<Long, Long>(categoryId,categoryId));

        }

        JavaPairRDD<Long,Long> top10CategoryIdRDD = sc.parallelizePairs(top10CategoryIdList);

        /**
         * 第二步:计算top10品类被各session点击的次数
         */
        JavaPairRDD<String, Iterable<Row>> sessionId2DetailsRDD = sessionId2DetailRDD.groupByKey();

        JavaPairRDD<Long, String> categoryId2SessionCountRDD = sessionId2DetailsRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Iterable<Row>>, Long, String>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<Long, String>> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
                String sessionId = tuple._1;
                Iterator<Row> iterator = tuple._2.iterator();
                Map<Long, Long> categoryCountMap = new HashMap<>();

                //计算出该session对每个品类的点击次数
                while (iterator.hasNext()) {
                    Row row = iterator.next();
                    if (row.get(6) != null) {
                        long categoryId = row.getLong(6);
                        Long count = categoryCountMap.get(categoryId);
                        if (count == null) {
                            count = 0L;
                        }
                        count++;
                        categoryCountMap.put(categoryId, count);
                    }
                }

                // 返回结果，<categoryId,sessionId,count>格式
                List<Tuple2<Long, String>> list = new ArrayList<>();

                for (Map.Entry<Long, Long> categoryCountEntry : categoryCountMap.entrySet()) {
                    long categoryId = categoryCountEntry.getKey();
                    long count = categoryCountEntry.getValue();
                    String value = sessionId + "," + count;
                    list.add(new Tuple2<Long, String>(categoryId, value));
                }
                return list;
            }
        });

        //获取到top10热门品类被各个session点击的次数
        JavaPairRDD<Long, String> top10CategorySessionCountRDD = top10CategoryIdRDD.join(categoryId2SessionCountRDD).mapToPair(new PairFunction<Tuple2<Long, Tuple2<Long, String>>, Long, String>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Long, String> call(Tuple2<Long, Tuple2<Long, String>> tuple) throws Exception {
                return new Tuple2<Long, String>(tuple._1, tuple._2._2);
            }
        });

        /**
         * 第三步: 分组取TopN算法实现,获取每个品类的top10活跃用户
          */
        JavaPairRDD<Long, Iterable<String>> top10CategorySessionCountsRDD = top10CategorySessionCountRDD.groupByKey();

        JavaPairRDD<String, String> top10SessionRDD = top10CategorySessionCountsRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<Long, Iterable<String>>, String, String>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Iterable<Tuple2<String, String>> call(Tuple2<Long, Iterable<String>> tuple) throws Exception {

                long categoryId = tuple._1;
                Iterator<String> iterator = tuple._2.iterator();

                //定义取topN的排序数组
                String[] top10Sessions = new String[10];
                while (iterator.hasNext()){

                    String sessionCount = iterator.next();
                    long count = Long.valueOf(sessionCount.split(",")[1]);
                    //遍历排序数组
                    int top10SessionsLength = top10Sessions.length;
                    for (int i = 0; i < top10SessionsLength; i++) {
                        //如果当前i为没有数据,那么直接将i位数据赋值为当前sessionCount
                        if (top10Sessions[i]==null){
                            top10Sessions[i] = sessionCount;
                            break;
                        }else {
                            long _count = Long.valueOf(top10Sessions[i].split(",")[1]);

                            //如果sessionCount比i位的sessionCount要大
                            if (count>_count){
                                //从排序数组最后一位开始到i位,所有数据往后挪一位
                                for (int j=9;j>i;j--){
                                    top10Sessions[j] = top10Sessions[j-1];
                                }
                                //将i位赋值为sessionCount
                                top10Sessions[i] = sessionCount;
                                break;
                            }
                            //比较小,继续外层循环
                        }
                    }
                }

                //将数据写入mysql表中
                List<Tuple2<String,String>> list = new ArrayList<>();

                for (String sessionCount : top10Sessions) {
                    if (StringUtils.isNotEmpty(sessionCount)){
                        String sessionId = sessionCount.split(",")[0];
                        long count = Long.valueOf(sessionCount.split(",")[1]);
                        //将top10session插入mysql表中
                        Top10Session top10Session = new Top10Session(taskId,categoryId,sessionId,count);

                        ITop10SessionDAO top10SessionDAO = DAOFactory.getTop10SessionDAO();
                        top10SessionDAO.insert(top10Session);
                        //放入list
                        list.add(new Tuple2<String, String>(sessionId,sessionId));
                    }
                }
                return list;
            }
        });

        /**
         * 第四步:获取top10活跃session明细数据写入mysql
         */
        JavaPairRDD<String, Tuple2<String, Row>> extractSessionDetailRDD = top10SessionRDD.join(sessionId2DetailRDD);

       /* extractSessionDetailRDD.foreach(new VoidFunction<Tuple2<String, Tuple2<String, Row>>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public void call(Tuple2<String, Tuple2<String, Row>> tuple) throws Exception {
                Row row = tuple._2._2;
                SessionDetail sessionDetail = new SessionDetail();
                sessionDetail.setTaskId(taskId);
                sessionDetail.setUserId(row.getLong(1));
                sessionDetail.setSessionId(row.getString(2));
                sessionDetail.setPageId(row.getLong(3));
                sessionDetail.setActionTime(row.getString(4));
                sessionDetail.setSearchKeyword(row.getString(5));
                sessionDetail.setClickCategoryId(row.getLong(6));
                sessionDetail.setClickProductId(row.getLong(7));
                sessionDetail.setOrderCategoryIds(row.getString(8));
                sessionDetail.setOrderProductIds(row.getString(9));
                sessionDetail.setPayCategoryIds(row.getString(10));
                sessionDetail.setPayProductIds(row.getString(11));

                ISessionDetailDAO sessionDetailDAO = DAOFactory.getSessionDetailDAO();
                sessionDetailDAO.insert(sessionDetail);
            }
        });*/

        extractSessionDetailRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Tuple2<String, Row>>>>() {
            @Override
            public void call(Iterator<Tuple2<String, Tuple2<String, Row>>> iterator) throws Exception {
                List<SessionDetail> sessionDetails = new ArrayList<>();
                while (iterator.hasNext()){
                    Tuple2<String,Tuple2<String,Row>> tuple = iterator.next();
                    Row row = tuple._2._2;
                    SessionDetail sessionDetail = new SessionDetail();
                    sessionDetail.setTaskId(taskId);
                    sessionDetail.setUserId(row.getLong(1));
                    sessionDetail.setSessionId(row.getString(2));
                    sessionDetail.setPageId(row.getLong(3));
                    sessionDetail.setActionTime(row.getString(4));
                    sessionDetail.setSearchKeyword(row.getString(5));
                    sessionDetail.setClickCategoryId(row.getLong(6));
                    sessionDetail.setClickProductId(row.getLong(7));
                    sessionDetail.setOrderCategoryIds(row.getString(8));
                    sessionDetail.setOrderProductIds(row.getString(9));
                    sessionDetail.setPayCategoryIds(row.getString(10));
                    sessionDetail.setPayProductIds(row.getString(11));

                    sessionDetails.add(sessionDetail);
                }
                ISessionDetailDAO sessionDetailDAO = DAOFactory.getSessionDetailDAO();
                sessionDetailDAO.insertBatch(sessionDetails);
            }
        });



    }

}
